import { Injectable } from '@angular/core';
import { Observable } from 'rxjs/Observable';
import { Subject } from 'rxjs/Subject';
import { Observer } from "rxjs/Rx";
import { GUID } from "../guid";
import { WebsocketHelpers } from './websocket-helpers';
/**
 * WebSocket服务对象
 * 
 * @export
 * @class WebSocketService
 */
@Injectable()
export class WebSocketService {
  /**
  * guid
  * 唯一的id标示
  * @type {string}
  * @memberof WebSocketService
  */
  public guid: string = GUID.getNewGUIDString();

  private send4Mode: WebSocketSendMode = WebSocketSendMode.Observable;
  /**
   * 重连次数
   * 
   * @private
   * 
   * @memberof WebSocketService
   */
  private reconnectAttempts = 0;
  /**
   * 发送给服务端的消息队列
   * 
   * @private
   * 
   * @memberof WebSocketService
   */
  private sendQueue: any[] = [];
  /**
   * 监听WebSocket连接打开的回调集合
   * 
   * @private
   * 
   * @memberof WebSocketService
   */
  private onOpenCallbacks: any[] = [];
  /**
   * 监听WebSocket发送消息的回调集合
   * 
   * @private
   * 
   * @memberof WebSocketService
   */
  private onMessageCallbacks: any[] = [];
  /**
   * 监听WebSocket通道出现错误的回调集合
   * 
   * @private
   * 
   * @memberof WebSocketService
   */
  private onErrorCallbacks: any[] = [];
  /**
   * 监听WebSocket通道关闭的回调集合
   * 
   * @private
   * 
   * @memberof WebSocketService
   */
  private onCloseCallbacks: any[] = [];
  /**
   * WebSocket 连接状态
   * 
   * @private
   * 
   * @memberof WebSocketService
   */
  private readyStateConstants = {
    // 连接中
    'CONNECTING': 0,
    // 已连接
    'OPEN': 1,
    // 关闭中
    'CLOSING': 2,
    // 已关闭
    'CLOSED': 3,
    // 重连失败
    'RECONNECT_ABORTED': 4,
  };
  /**
   * 发送给WebSocket服务器正常关闭的状态码
   * @private
   * @memberof WebSocketService
   */
  private normalCloseCode = 1000;
  /**
   * 发送给WebSocket服务器重连状态码集合
   * @private
   * @memberof WebSocketService
   */
  private reconnectableStatusCodes = [4000];
  /**
   * 当前WebSocket对象
   * @private
   * @type {WebSocket}
   * @memberof WebSocketService
   */
  public socket: WebSocket;
  /**
   * 可订阅的数据流对象
   * @private
   * @type {Subject<any>}
   * @memberof WebSocketService
   */
  private dataStream: Subject<any>;
  /**
   * 可订阅的错误信息流对象
   * @private
   * @type {Subject<any>}
   * @memberof WebSocketService
   */
  private errorMessages: Subject<any>;
  /**
   * 内部连接状态
   * @private
   * @type {number}
   * @memberof WebSocketService
   */
  private internalConnectionState: number;

  /**
   * Creates an instance of WebSocketService.
   * @param {string} url ws://192.168.0.1/webSocketService || wss://192.168.0.1/webSocketService
   * @param {Array<string>} [protocols] WebSocket服务器可以理解的通信协议（xmpp，Stomp)
   * @param {WebSocketConfig} [config] WebSocket 配置
   * @param {BinaryType} [binaryType] 协商数据类型
   * @memberof WebSocketService
   */
  constructor(
    private url: string,
    private protocols?: Array<string>,
    private config?: WebSocketConfig,
    private binaryType?: BinaryType, ) {
    let match = new RegExp('wss?:\/\/').test(url);
    if (!match) {
      throw new Error('Invalid url provided');
    }
    this.config = config || { initialTimeout: 500, maxTimeout: 300000, reconnectIfNotNormalClose: false };
    this.binaryType = binaryType || "blob";
    this.dataStream = new Subject();
    this.errorMessages = new Subject();
    this.connect(true);
  }

  /**
   * 
   * 
   * @param {boolean} [force=false] 
   * 
   * @memberof WebSocketService
   */
  connect(force = false) {
    // console.log("WebSocket connecting...");
    let self = this;

    if (force || !this.socket || this.socket.readyState !== this.readyStateConstants.OPEN) {
      self.socket = this.protocols ? new WebSocket(this.url, this.protocols) : new WebSocket(this.url);
      self.socket.binaryType = self.binaryType.toString();
      self.socket.onopen = (ev: Event) => {
        // console.log('onOpen: ', ev);
        this.onOpenHandler(ev);
      };
      self.socket.onmessage = (ev: MessageEvent) => {
        // console.log('onNext: ', ev.data);
        self.onMessageHandler(ev);
        this.dataStream.next(ev);
      };
      this.socket.onclose = (ev: CloseEvent) => {
        // console.log('onClose ', ev);
        self.onCloseHandler(ev);
      };
      this.socket.onerror = (ev: ErrorEvent) => {
        // console.log('onError ', ev);
        self.onErrorHandler(ev);
        this.errorMessages.next(ev);
        this.socket.close();
        //this._message.info('colse' + ev, { nzDuration: 3000 });
        //console.log('close ', ev);
      };

    }


  }
  onOpenHandler(event: Event) {
    this.reconnectAttempts = 0;
    this.notifyOpenCallbacks(event);
    this.fireQueue();
  }
  onMessageHandler(message: MessageEvent) {
    let self = this;
    let currentCallback;
    for (let i = 0; i < self.onMessageCallbacks.length; i++) {
      currentCallback = self.onMessageCallbacks[i];
      currentCallback.fn.apply(self, [message]);
    }
  }
  onCloseHandler(event: CloseEvent) {
    this.notifyCloseCallbacks(event);
    if ((this.config.reconnectIfNotNormalClose && event.code !== this.normalCloseCode)
      || this.reconnectableStatusCodes.indexOf(event.code) > -1) {
      this.reconnect();
    } else {
      this.sendQueue = [];
      this.dataStream.complete();
    }
  }
  onErrorHandler(event: Event) {
    this.notifyErrorCallbacks(event);
  }
  /**
   * 通知连接打开回调
   * 
   * @param {any} event 
   * 
   * @memberof WebSocketService
   */
  notifyOpenCallbacks(event: Event) {
    for (let i = 0; i < this.onOpenCallbacks.length; i++) {
      this.onOpenCallbacks[i].call(this, event);
    }
  }
  notifyCloseCallbacks(event: CloseEvent) {
    for (let i = 0; i < this.onCloseCallbacks.length; i++) {
      this.onCloseCallbacks[i].call(this, event);
    }
  }
  notifyErrorCallbacks(event: Event) {
    for (let i = 0; i < this.onErrorCallbacks.length; i++) {
      this.onErrorCallbacks[i].call(this, event);
    }
  }
  getErrorStream(): Subject<any> {
    return this.errorMessages;
  }


  /**
   * Set send(data) function return mode
   * @param mode
   */
  private setSend4Mode(mode: WebSocketSendMode): void {
    this.send4Mode = mode;
  }
  /**
   * Run in Block Mode
   * Return true when can send and false in socket closed
   * @param data
   * @returns {boolean}
   */
  private send4Direct(data: any, binary?: boolean): boolean {
    let self = this;
    if (this.getReadyState() !== this.readyStateConstants.OPEN
      && this.getReadyState() !== this.readyStateConstants.CONNECTING) {
      this.connect();
    }
    self.sendQueue.push({ message: data, binary: binary });
    if (self.socket.readyState === self.readyStateConstants.OPEN) {
      self.fireQueue();
      return true;
    } else {
      return false;
    }
  }
  /**
   * Return Promise
   * When can Send will resolve Promise
   * When Socket closed will reject Promise
   * @param data
   * @returns {Promise<any>}
   */
  private send4Promise(data: any, binary?: boolean): Promise<any> {
    return new Promise(
      (resolve, reject) => {
        if (this.send4Direct(data, binary)) {
          return resolve();
        } else {
          return reject(Error('Socket connection has been closed'));
        }
      }
    )
  }
  /**
   * Return cold Observable
   * When can Send will complete observer
   * When Socket closed will error observer
   * @param data
   * @returns {Observable<any>}
   */
  private send4Observable(data: any, binary?: boolean): Observable<any> {
    return Observable.create((observer: Observer<any>) => {
      if (this.send4Direct(data, binary)) {
        return observer.complete();
      } else {
        return observer.error('Socket connection has been closed');
      }
    });
  }
  /**
   * Use {mode} mode to send {data} data
   * If no specify, Default SendMode is Observable mode
   * @param data
   * @param mode
   * @param binary
   * @returns {any}
   */
  send(data: any, mode?: WebSocketSendMode, binary?: boolean): any {
    switch (typeof mode !== "undefined" ? mode : this.send4Mode) {
      case WebSocketSendMode.Direct:
        return this.send4Direct(data, binary);
      case WebSocketSendMode.Promise:
        return this.send4Promise(data, binary);
      case WebSocketSendMode.Observable:
        return this.send4Observable(data, binary);
      default:
        throw Error("WebSocketSendMode Error.");
    }
  }

  getDataStream(): Subject<any> {
    return this.dataStream;
  }

  fireQueue() {
    // console.log("fireQueue()");
    while (this.sendQueue.length && this.socket.readyState === this.readyStateConstants.OPEN) {
      let data = this.sendQueue.shift();
      // console.log("fireQueue: ", data);
      if (data.binary) {
        this.socket.send(data.message);
      } else {
        this.socket.send(

          WebsocketHelpers.isString(data.message) ? data.message : JSON.stringify(data.message)
        );
      }
      // data.deferred.resolve();
    }
  }

  onOpen(cb: any) {
    this.onOpenCallbacks.push(cb);
    return this;
  };

  onClose(cb: any) {
    this.onCloseCallbacks.push(cb);
    return this;
  }

  onError(cb: any) {
    this.onErrorCallbacks.push(cb);
    return this;
  };

  onMessage(callback: any, options?: any) {
    if (!WebsocketHelpers.isFunction(callback)) {
      throw new Error('Callback must be a function');
    }
    this.onMessageCallbacks.push({
      fn: callback,
      pattern: options ? options.filter : undefined,
      autoApply: options ? options.autoApply : true
    });
    return this;
  }

  reconnect() {
    this.close(true);
    let backoffDelay = this.getBackoffDelay(++this.reconnectAttempts);
    // let backoffDelaySeconds = backoffDelay / 1000;
    // console.log('Reconnecting in ' + backoffDelaySeconds + ' seconds');
    setTimeout(() => this.connect(), backoffDelay);
    return this;
  }

  close(force: boolean = false) {
    if (force || !this.socket.bufferedAmount) {
      this.socket.close(this.normalCloseCode);
    }
    return this;
  };

  // Exponential Backoff Formula by Prof. Douglas Thain
  // http://dthain.blogspot.co.uk/2009/02/exponential-backoff-in-distributed.html
  getBackoffDelay(attempt: number) {
    let R = Math.random() + 1;
    let T = this.config.initialTimeout;
    let F = 2;
    let N = attempt;
    let M = this.config.maxTimeout;
    return Math.floor(Math.min(R * T * Math.pow(F, N), M));
  };

  setInternalState(state: number) {
    if (Math.floor(state) !== state || state < 0 || state > 4) {
      throw new Error('state must be an integer between 0 and 4, got: ' + state);
    }
    this.internalConnectionState = state;
  }

  /**
   * Could be -1 if not initzialized yet
   * @returns {number}
   */
  getReadyState() {
    if (this.socket == null) {
      return -1;
    }
    return this.internalConnectionState || this.socket.readyState;
  }
}

export interface WebSocketConfig {
  initialTimeout: number;
  maxTimeout: number;
  reconnectIfNotNormalClose: boolean;
}

export enum WebSocketSendMode {
  Direct, Promise, Observable
}

export type BinaryType = "blob" | "arraybuffer";


// WEBPACK FOOTER //
// ./src/app/service/data-source/websocket.service.ts